package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dwd.log.job.JtpLogClickHouse;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author: FuHe
 * @Date: 2025/5/18
 */
public class LogSplitProcessFuncation extends ProcessFunction<String, String> {
    /**
     * 测输出流输出时标记OutputTag实例
     */
    private final OutputTag<String> errorTag;
    private final OutputTag<String> stratTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public LogSplitProcessFuncation(OutputTag<String> errorTag, OutputTag<String> stratTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.stratTag = stratTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        // 第一步、解析JSON数据
        JSONObject jsonObject = JSON.parseObject(value);
        // 第二步、错误日志
        if (null != jsonObject.get("err")) {
            // 侧边流输出
            ctx.output(errorTag, value);
            // 错误数据已经单独保存，无需在保存，直接删除
            jsonObject.remove("err");
        }
        // 第三部、启动日志
        if (null != jsonObject.get("start")) {
            //侧变流输出
            ctx.output(stratTag, jsonObject.toJSONString());
        }
        // 第四步、页面日志
        else {
            // 获取曝光和动作数据
            JSONArray displaysJsonArray = jsonObject.getJSONArray("displays");
            JSONArray actionJsonArray = jsonObject.getJSONArray("actions");
            // 删除曝光和动作
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            //输出页面日志
            out.collect(jsonObject.toJSONString());
            //第五步、曝光日志
            if (null != displaysJsonArray && !displaysJsonArray.isEmpty()) {
                for (int i = 0; i < displaysJsonArray.size(); i++) {
                    JSONObject displaysJsonObject = displaysJsonArray.getJSONObject(i);
                    // 添加字段值
                    jsonObject.put("display", displaysJsonObject);
                    // 输出
                    ctx.output(displayTag, jsonObject.toJSONString());
                }
            }
            // 将上述添加曝光数据字段值删除
            jsonObject.remove("display");
            // 第六步、动作日志
            if (null != actionJsonArray && !actionJsonArray.isEmpty()){
                for (int i = 0; i < actionJsonArray.size(); i++) {
                    JSONObject actionJsonObject = actionJsonArray.getJSONObject(i);
                    // 添加字段值
                    jsonObject.put("action", actionJsonObject);
                    // 输出
                    ctx.output(actionTag, jsonObject.toJSONString());
                }
            }
        }
    }
}
